package com.nulldev.util.concurrency.threadIt.v4.emapi.cntft;

import static com.nulldev.util.concurrency.threadIt.v4.emapi.uft.ExecutableUtils.callAsyncEventsOrFail;
import static com.nulldev.util.concurrency.threadIt.v4.emapi.uft.ExecutableUtils.callAsyncFails;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import com.nulldev.util.JVM.Arguments;
import com.nulldev.util.VariableAPI.MathUtil;
import com.nulldev.util.concurrency.threadIt.v3.extensions.TaskStatus;
import com.nulldev.util.concurrency.threadIt.v4.asyncTasks.AsyncTaskCallback;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutorManager;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.concurrency.threadIt.v4.locking.ResettableCountDownLatch;
import com.nulldev.util.data.Arrays.queues.EnhancedLinkedBlockingQueue;
import com.nulldev.util.internal.backport.concurrency9.Lists;
import com.nulldev.util.io.FastIO.FastIO;
import com.nulldev.util.manage.nullUtil;
import com.nulldev.util.manage.legacy.Timer;

public class ControllableFutureTask<T> implements Executable<T>, Runnable {

	private final static boolean ENABLE_OFFLOADED_ASYNC = Arguments.notArguments("--Tasks.disableAsyncOffload", "--ControllableFutureTask.disableAsyncOffload");
	private final static boolean ENABLE_ELBQ = Arguments.hasArgument("--Tasks.enableELBQ");
	private final static boolean USE_NEW_FAIL_LOGIC = Arguments.notArguments("--Tasks.disableNewFailureLogic");
	private final static boolean CFT_USE_CORRECT_LOGIC = true;

	private final ResettableCountDownLatch countDownLatch = new ResettableCountDownLatch(1);
	private final AtomicReference<T> result = new AtomicReference<T>();
	private final AtomicReference<TaskStatus> status = new AtomicReference<TaskStatus>(TaskStatus.RUNNING);
	private final AtomicReference<Throwable> cachedError = new AtomicReference<Throwable>();
	private final BlockingQueue<AsyncTaskCallback<T>> callback;

	public ControllableFutureTask() {
		this.callback = ENABLE_ELBQ ? new EnhancedLinkedBlockingQueue<AsyncTaskCallback<T>>() : new LinkedTransferQueue<AsyncTaskCallback<T>>();
	}

	public ControllableFutureTask(final BlockingQueue<AsyncTaskCallback<T>> callback) {
		this.callback = callback;
	}

	@Override
	public T run_st() {
		if (CFT_USE_CORRECT_LOGIC) {
			return this.waitTillCompletion();
		} else {
			return this.result.get();
		}
	}

	@Override
	public T waitTillCompletion() {
		if (this.completed()) {
			if (this.status.get() == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC)
				throw new RuntimeException(this.cachedError.get());
			else if (this.status.get() == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC)
				throw new RuntimeException("Submission was rejected.");
			else
				return this.result.get();
		} else {
			try {
				this.countDownLatch.await();
			} catch (InterruptedException e) {
			}
			if (this.status.get() == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC)
				throw new RuntimeException(this.cachedError.get());
			else if (this.status.get() == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC)
				throw new RuntimeException("Submission was rejected.");
			else
				return this.result.get();
		}
	}

	@Override
	public T waitFor(final long amount, final TimeUnit millis) {
		if (amount < 0 || millis == null)
			throw new IllegalArgumentException("amount < 0!");
		if (this.completed())
			if (this.status.get() == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC)
				throw new RuntimeException(this.cachedError.get());
			else if (this.status.get() == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC)
				throw new RuntimeException("Submission was rejected.");
			else
				return this.result.get();
		try {
			this.countDownLatch.await(amount, millis);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
		if (this.status.get() == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException(this.cachedError.get());
		else if (this.status.get() == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException("Submission was rejected.");
		else
			return this.result.get();
	}

	@Override
	public boolean addCallbackTask(final AsyncTaskCallback<T> task) {
		if (task == null) {
			return false;
		} else if (this.completed()) {

			if (this.status.get() == TaskStatus.FAILED) {
				task.onError(this.cachedError.get());
			} else if (this.status.get() == TaskStatus.SUBMISSION_FAILED) {
				task.onSubmissionRejection();
			} else {
				try {
					task.onCompletion(this.result.get());
				} catch (Throwable tx) {
					task.onError(tx);
				}
			}
			return false;

		} else {
			try {
				return this.callback.offer(task);
			} catch (Throwable _internal) {

				if (this.completed()) {

					if (this.status.get() == TaskStatus.FAILED) {
						task.onError(this.cachedError.get());
					} else if (this.status.get() == TaskStatus.SUBMISSION_FAILED) {
						task.onSubmissionRejection();
					} else {
						task.onCompletion(this.result.get());
					}

				} else {
					FastIO.get().printlnError(_internal);
				}

				return false;
			}
		}
	}

	@Override
	public List<AsyncTaskCallback<?>> callbacks() {
		return Lists.of(this.callback.toArray(new AsyncTaskCallback[this.callback.size()]));
	}

	@Override
	public T result() {
		return this.result.get();
	}

	@Override
	public boolean completed() {
		return this.status.get() == TaskStatus.COMPLETED || this.status.get() == TaskStatus.FAILED || this.status.get() == TaskStatus.KILLED
				|| this.status.get() == TaskStatus.CANCELLED || this.status.get() == TaskStatus.SUBMISSION_FAILED;
	}

	public boolean complete(final T value) {
		if (this.status.get() == TaskStatus.KILLED || this.status.get() == TaskStatus.FAILED || this.status.get() == TaskStatus.CANCELLED)
			return false;
		if (this.status.get() == TaskStatus.COMPLETED) {
			System.out.println("[!] [ControllableFutureTask] Task already completed, something isn't right...");
			if (nullUtil.IS_NON_MAINLINE_BUILD)
				new Exception("[ControllableFutureTask] More than one invocation of complete(T).").printStackTrace();
			return false;
		}
		this.result.lazySet(value);
		this.status.set(TaskStatus.COMPLETED);
		// FIXME: Setting ENABLE_OFFLOADED_ASYNC below breaks things.
		callAsyncEventsOrFail(false, false, this.callback, value);
		this.countDownLatch.countDown();
		return true;
	}

	public void fail(final Throwable e) {
		if (this.status.get() == TaskStatus.KILLED || this.status.get() == TaskStatus.COMPLETED || this.status.get() == TaskStatus.CANCELLED)
			return;
		this.status.set(TaskStatus.FAILED);
		this.cachedError.lazySet(e);
		// FIXME: Setting ENABLE_OFFLOADED_ASYNC below breaks things.
		callAsyncFails(false, false, this.callback, e);
		this.countDownLatch.countDown();
		if (!USE_NEW_FAIL_LOGIC)
			throw new RuntimeException(e);
	}

	@Override
	public void kill() {
		if (this.completed())
			return;
		this.status.lazySet(TaskStatus.KILLED);
		this.countDownLatch.countDown();
	}

	public void cancel(final boolean interrupt) {
		this.kill();
	}

	public void reset() {
		this.countDownLatch.countDownThenReset();
		this.cachedError.lazySet(null);
		this.result.lazySet(null);
		this.status.lazySet(TaskStatus.RUNNING);
	}

	@Override
	public void rejectSubmission() {
		/* Unsupported, no-op */
	}

	@Override
	public Object getFlag(final int flag) {
		switch (flag) {
			case FT_SUPPORTS_POLLING:
				return FT_TRUE;
			default:
				return FT_NOT_SUPPORTED;
		}
	}

	@Override
	public boolean hasTrace() {
		return false;
	}

	@Override
	public int getFlags() {
		return FT_SUPPORTS_POLLING;
	}

	@Override
	public void signalQueued() {
		if (!this.callback.isEmpty()) {
			if (ENABLE_OFFLOADED_ASYNC) {
				this.callback.iterator().forEachRemaining((_o) -> {
					if (_o == null)
						return;
					final AsyncTaskCallback<T> o = _o;
					ExecutorManager.get().autoAsyncSubmit(true, () -> {
						try {
							o.onQueued();
						} catch (Throwable ec) {
							ec.printStackTrace(System.err);
							System.err.flush();
						}
					});
				});
			} else {
				this.callback.iterator().forEachRemaining((o) -> {
					if (o == null)
						return;
					try {
						o.onQueued();
					} catch (Throwable ec) {
						ec.printStackTrace(System.err);
						System.err.flush();
					}
				});
			}
		}
	}

	@Override
	public TaskStatus status() {
		return this.status.get();
	}

	public boolean isRunning() {
		return this.status.get() == TaskStatus.RUNNING;
	}

	@Override
	public Throwable getException() {
		return this.cachedError.get();
	}

	@Override
	public void run() {
		throw new UnsupportedOperationException("Attempted to run a ControllableFutureTask, this is not supported.");
	}

	@Override
	public String toString() {
		return "ControllableFutureTask[hash=" + MathUtil.toHex(super.hashCode()) + ",state=" + this.status + ",result=" + this.result + "]";
	}

	/* FT 1.1: Polling */

	@Override
	public T pollUntilCompletion(final int polling_algorithm) {
		switch (polling_algorithm) {
			case FT_POLLING_IMMEDIATE: {
				while (this.isRunning()) {
					try {
						Thread.sleep(1L);
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
				}
				break;
			}
			case FT_POLLING_NEW_SCALING: {
				// TODO: Implement and create an algorithm for FT_POLLING_NEW_SCALING.
			}
			default:
			case FT_POLLING_LEGACY_SCALING: {
				final int limit = 8;
				int waitingInt = limit;
				final int s = (int) (FT_POLLING_LEGACY_SCALING_WAITING_INTERVAL / limit);
				while (this.isRunning()) {
					if (waitingInt <= 0) {
						waitingInt = limit + 1;
					}
					try {
						Thread.sleep((long) Math.ceil(s / waitingInt));
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
					waitingInt -= 1;
				}
				break;
			}
		}

		if (this.status.get() == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException(this.cachedError.get());
		else if (this.status.get() == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException("Submission was rejected.");
		else
			return this.result.get();
	}

	@Override
	public T pollFor(final long amount, final TimeUnit unit, final int polling_algorithm) {
		final Timer time = new Timer();
		time.start(ExecutorManager.get().executor());
		final long am = unit.toMillis(amount);
		switch (polling_algorithm) {
			case FT_POLLING_IMMEDIATE: {
				while (this.isRunning() && !time.hasReached(am)) {
					try {
						Thread.sleep(1L);
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
				}
				break;
			}
			case FT_POLLING_NEW_SCALING: {
				// TODO: Implement and create an algorithm for FT_POLLING_NEW_SCALING.
			}
			default:
			case FT_POLLING_LEGACY_SCALING: {
				final int limit = 8;
				int waitingInt = limit;
				final int s = (int) (FT_POLLING_LEGACY_SCALING_WAITING_INTERVAL / limit);
				while (this.isRunning() && !time.hasReached(am)) {
					if (waitingInt <= 0) {
						waitingInt = limit + 1;
					}
					try {
						Thread.sleep((long) Math.ceil(s / waitingInt));
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
					waitingInt -= 1;
				}
				break;
			}
		}
		time.stop();

		if (this.status.get() == TaskStatus.FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException(this.cachedError.get());
		else if (this.status.get() == TaskStatus.SUBMISSION_FAILED && USE_NEW_FAIL_LOGIC)
			throw new RuntimeException("Submission was rejected.");
		else
			return this.result.get();
	}
}
